package xdu.lz.pc;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

class TaskQueue1<T> {

    private Queue<T> queue;

    private static final int DEFAULT_CAPACITY = 10;

    private final int MAX_CAPACITY;

    public TaskQueue1() {
        this(DEFAULT_CAPACITY);
    }

    public TaskQueue1(int capacity) {
        this.MAX_CAPACITY = capacity;
        this.queue = new LinkedList<>();
    }

    public  void offer(T e) throws InterruptedException {
        synchronized (queue) {
            while (queue.size() >= MAX_CAPACITY) {
                queue.wait();
            }
            queue.add(e);
            System.out.println(Thread.currentThread().getName()+"生产了"+e);
            queue.notifyAll();
        }
    }

    public T poll() throws InterruptedException {
        synchronized (queue) {
            while (queue.isEmpty()) {
                queue.wait();
            }
            queue.notifyAll();
            System.out.println(Thread.currentThread().getName()+"消费了"+queue.peek());
            return queue.poll();
        }
    }
}

class TaskQueue2<T> {

    private ReentrantLock lock = new ReentrantLock();

    private Condition condition = lock.newCondition();

    private Queue<T> queue;

    private static final int DEFAULT_CAPACITY = 100;

    private final int MAX_CAPACITY;

    public TaskQueue2() {
        this(DEFAULT_CAPACITY);
    }

    public TaskQueue2(int capacity) {
        this.MAX_CAPACITY = capacity;
        this.queue = new LinkedList<>();
    }

    public void offer(T e) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() >= MAX_CAPACITY) {
                condition.await();
            }
            queue.add(e);
            System.out.println(Thread.currentThread().getName()+"生产了"+e);
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public T poll() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                condition.await();
            }
            condition.signalAll();
            System.out.println(Thread.currentThread().getName()+"消费了"+queue.peek());
            return queue.poll();
        } finally {
            lock.unlock();
        }

    }
}


public class Code02_SimplePC2 {
    private AtomicInteger counter
            = new AtomicInteger(0);
    private TaskQueue1<Integer> queue
            = new TaskQueue1<>(3);

    public void produce() throws InterruptedException {
        while (true) {
            //如果这样写会有线程安全问题，counter.get会被 conter.getAndIncrement更新
//            System.out.println(Thread.currentThread().getName() + "生产了:" + (counter.incrementAndGet()));
//            queue.offer(counter.get());
            //这样不会有线程安全问题
            queue.offer(counter.getAndIncrement());
            Thread.sleep(20);
        }
    }

    public void consume() throws InterruptedException {
        while (true) {
            queue.poll();
            Thread.sleep(2000);
        }

    }

    public static void main(String[] args) {
        Code02_SimplePC2 pc = new Code02_SimplePC2();
        CountDownLatch downLatch = new CountDownLatch(1);
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    downLatch.await();
                    pc.produce();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
            new Thread(() -> {
                try {
                    pc.consume();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        downLatch.countDown();
    }
}
